package com.atguigu.champter5.Flink05.test;

import com.atguigu.champter5.Flink05.test.bean.AdsClickLog;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Flink_Ads_Provence_Count {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
       env.readTextFile("datas/input/AdClickLog.csv")
               .map(new MapFunction<String, AdsClickLog>() {
                   @Override
                   public AdsClickLog map(String value) throws Exception {
                       String[] data = value.split(",");
                       return new AdsClickLog(
                               Long.valueOf(data[0]),
                               Long.valueOf(data[1]),
                               String.valueOf(data[2]),
                               String.valueOf(data[3]),
                               Long.valueOf(data[4])
                       );
                   }
               })
               .map(new MapFunction<AdsClickLog, Tuple2<String,Integer>>() {
                   @Override
                   public Tuple2<String, Integer> map(AdsClickLog value) throws Exception {
                       return new Tuple2(value.getProvince() + ":" + value.getAdId(),1);
                   }
               })
               .keyBy(Tuple2->Tuple2.f0)
               .sum(1).print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
